package com.lagou.rpc.client.netty;

import com.lagou.rpc.client.Connection;
import com.lagou.rpc.client.ConnectionSource;
import com.lagou.rpc.client.netty.handler.RPCResponseDataRequestHandler;
import com.lagou.rpc.client.netty.loadbalance.FastResponseLoadbalancer;
import com.lagou.rpc.client.netty.loadbalance.RoundRobinLoadbalancer;
import com.lagou.rpc.common.converter.Converter;
import com.lagou.rpc.common.entity.HostPortMetaData;
import com.lagou.rpc.common.entity.RPCResponseData;
import com.lagou.rpc.common.exceptions.SimpleRpcException;
import com.lagou.rpc.common.handler.RequestHandlerRegistry;
import com.lagou.rpc.common.handler.SimpleRpcProtocolDecoder;
import com.lagou.rpc.common.handler.SimpleRpcProtocolEncoder;
import com.lagou.rpc.common.handler.SimpleRpcTransferDataHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;

/**
 * netty实现的连接源
 *
 * @author wlz
 * @date 2020/5/1
 */
public class NettyConnectionSource implements ConnectionSource {

    private static final Logger logger = LoggerFactory.getLogger(NettyConnectionSource.class);

    private int workerThreadSize;
    private Converter converter;
    private long timeout;
    private LoadBalanceStrategy loadBalanceStrategy;
    private RequestResultRegistry requestResultRegistry;
    private Loadbalancer loadbalancer;
    private Bootstrap bootstrap;
    private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    private final ReentrantReadWriteLock.ReadLock readLock = readWriteLock.readLock();
    private final ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();
    private Map<String, HostPortMetaData> availableServerMap;
    private ScheduledThreadPoolExecutor refreshAvailableHostTask;

    public NettyConnectionSource(NettyConnectionConfig nettyConnectionConfig) {
        workerThreadSize = nettyConnectionConfig.getWorkerThreadSize();
        converter = nettyConnectionConfig.getConverter();
        this.loadBalanceStrategy = nettyConnectionConfig.getLoadBalanceStrategy();
        // 采用策略进行负载均衡获取服务节点
        switch (loadBalanceStrategy) {
            case ROUND_ROBIN:
                loadbalancer = new RoundRobinLoadbalancer();
                break;
            case LONGEST_RESPONSE:
                loadbalancer = new FastResponseLoadbalancer();
                break;
            default:
                throw new SimpleRpcException("not supported strategy: " + loadBalanceStrategy.name());
        }
        timeout = nettyConnectionConfig.getTimeout();
        requestResultRegistry = new RequestResultRegistry();
        bootstrap = new Bootstrap();
        this.availableServerMap = initHostMap(nettyConnectionConfig.getAvailableServerList());
        refreshAvailableHostTask = new ScheduledThreadPoolExecutor(1, new DefaultThreadFactory("refreshAvailableHostTask"));
        refreshAvailableHostTask.scheduleAtFixedRate(this::refreshHostAvailable, 0, 5, TimeUnit.SECONDS);
        init();
    }

    private void refreshHostAvailable() {
        try {
            availableServerMap.values().forEach(host -> {
                host.setUnavailable(false);
            });
        } catch (Throwable e) {
            logger.error("refresh task occur error", e);
        }
    }

    private Map<String, HostPortMetaData> initHostMap(Set<HostPortMetaData> availableServerList) {
        if (availableServerList == null) {
            return new HashMap<>(5);
        }
        HashMap<String, HostPortMetaData> map = new HashMap<>(availableServerList.size());
        for (HostPortMetaData hostPortMetaData : availableServerList) {
            map.put(hostPortMetaData.getHost() + ":" + hostPortMetaData.getPort(), hostPortMetaData);
        }
        return map;
    }

    private void init() {
        // 创建工作线程池
        NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(workerThreadSize, (r) -> {
            Thread thread = new Thread(r);
            thread.setUncaughtExceptionHandler((t, e) -> e.printStackTrace());
            return thread;
        });
        // 注册结果处理器
        RequestHandlerRegistry requestHandlerRegistry = new RequestHandlerRegistry();
        requestHandlerRegistry.register(RPCResponseData.class, new RPCResponseDataRequestHandler(requestResultRegistry));
        // 初始化客户端bootStrap
        bootstrap.group(nioEventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline()
                                // 协议处理，将请求数据转换为transferData
                                .addLast(new SimpleRpcProtocolDecoder())
                                // response编码
                                .addLast(new SimpleRpcProtocolEncoder())
                                // 处理请求rpc逻辑
                                .addLast(new SimpleRpcTransferDataHandler(converter, requestHandlerRegistry));
                    }
                });

    }


    @Override
    public Connection getConnection() {
        // 通过负载均衡获取到一个可用的服务
        HostResult availableConnect = getAvailableConnect();
        ChannelFuture channelFuture = availableConnect.channelFuture;
        HostPortMetaData hostAndPort = availableConnect.hostAndPort;
        Channel channel = channelFuture.channel();
        return new NettyConnection(channel, converter, hostAndPort, requestResultRegistry, timeout);
    }


    private HostResult getAvailableConnect() {
        HostPortMetaData hostAndPort = loadBalanceGetHost(loadbalancer);
        try {
            ChannelFuture channelFuture = bootstrap.connect(hostAndPort.getHost(), hostAndPort.getPort()).sync();
            return new HostResult(channelFuture, hostAndPort);
        } catch (Throwable e) {
            hostAndPort.setUnavailable(true);
            return getAvailableConnect();
        }
    }

    private static class HostResult {
        private ChannelFuture channelFuture;
        private HostPortMetaData hostAndPort;

        public HostResult(ChannelFuture channelFuture, HostPortMetaData hostAndPort) {
            this.channelFuture = channelFuture;
            this.hostAndPort = hostAndPort;
        }
    }


    private HostPortMetaData loadBalanceGetHost(Loadbalancer loadbalancer) {
        readLock.lock();
        try {
            if (availableServerMap.isEmpty()) {
                throw new SimpleRpcException("no available server!");
            }
            List<HostPortMetaData> availableHostList = availableServerMap.values().stream()
                    .filter(host -> !host.isUnavailable())
                    .collect(Collectors.toList());
            if (availableHostList.isEmpty()) {
                throw new SimpleRpcException("no available server!");
            }
            return loadbalancer.balance(new ArrayList<>(availableHostList));
        } finally {
            readLock.unlock();
        }
    }

    public void putHost(HostPortMetaData hostPortMetaData) {
        writeLock.lock();
        try {
            availableServerMap.put(hostPortMetaData.getHost() + ":" + hostPortMetaData.getPort(), hostPortMetaData);
        } finally {
            writeLock.unlock();
        }
    }

    public void removeHost(HostPortMetaData hostPortMetaData) {
        writeLock.lock();
        try {
            availableServerMap.remove(hostPortMetaData.getHost() + ":" + hostPortMetaData.getPort());
        } finally {
            writeLock.unlock();
        }
    }

    public int getWorkerThreadSize() {
        return workerThreadSize;
    }

    public void setWorkerThreadSize(int workerThreadSize) {
        this.workerThreadSize = workerThreadSize;
    }

    public Converter getConverter() {
        return converter;
    }

    public void setConverter(Converter converter) {
        this.converter = converter;
    }

    public long getTimeout() {
        return timeout;
    }

    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }

    public LoadBalanceStrategy getLoadBalanceStrategy() {
        return loadBalanceStrategy;
    }

    public void setLoadBalanceStrategy(LoadBalanceStrategy loadBalanceStrategy) {
        this.loadBalanceStrategy = loadBalanceStrategy;
    }

    public Map<String, HostPortMetaData> getAvailableServerMap() {
        return availableServerMap;
    }

    public void setAvailableServerMap(Map<String, HostPortMetaData> availableServerMap) {
        this.availableServerMap = availableServerMap;
    }
}
